package com.sunwayworld.basemodule.common.eptstorage.acqconfig.net;//package com.sunwayworld.basemodule.common.eptstorage.acqconfig.net;
//
//import com.sunwayworld.basemodule.common.eptstorage.acqconfig.bean.LimsEquiptAcqConfigInstBean;
//import com.sunwayworld.basemodule.common.eptstorage.acqconfig.service.LimsEquiptAcqConfigInstService;
//import com.sunwayworld.basemodule.common.eptstorage.equiptacqschedule.bean.LimsEquiptAcqScheduleTaskBean;
//import com.sunwayworld.basemodule.common.eptstorage.equiptacqschedule.service.impl.LimsEquiptAcqScheduleTaskServiceImpl;
//import com.sunwayworld.basemodule.common.eptstorage.util.Util;
//import com.sunwayworld.cloud.module.equipt.acqconfig.bean.CoreEquiptAcqConfigBean;
//import com.sunwayworld.framework.cache.redis.RedisHelper;
//import com.sunwayworld.framework.constant.Constant;
//import com.sunwayworld.framework.context.ApplicationContextHelper;
//import com.sunwayworld.framework.exception.InvalidDataException;
//import com.sunwayworld.framework.mybatis.mapper.MatchPattern;
//import com.sunwayworld.framework.mybatis.mapper.SearchFilter;
//import com.sunwayworld.framework.utils.ObjectUtils;
//import com.sunwayworld.framework.utils.StringUtils;
//import io.netty.bootstrap.Bootstrap;
//import io.netty.buffer.ByteBuf;
//import io.netty.channel.*;
//import io.netty.channel.nio.NioEventLoopGroup;
//import io.netty.channel.socket.nio.NioSocketChannel;
//import org.eclipse.paho.client.mqttv3.*;
//import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
//
//import java.nio.charset.StandardCharsets;
//import java.time.LocalDateTime;
//import java.util.*;
//import java.util.concurrent.CountDownLatch;
//import java.util.concurrent.TimeUnit;
//import java.util.stream.Collectors;
//
//public class NetEptMqttSender extends NetEptSender {
//
//    public MqttClient client;
//    private Map<String, Object> msg = null;
//    private Map<String,LimsEquiptAcqConfigInstBean> instMap = null;
//    private CountDownLatch waitConnectLatch = null;
//    private int reconnectTimes = 3;
//    public Object getMsg() {
//        return msg;
//    }
//    public void setMsg(String topic,Object value) {
//        this.msg.put(topic,value);
//    }
//
//    public NetEptMqttSender(CoreEquiptAcqConfigBean eptAcqBean) {
//        super(eptAcqBean);
//    }
//
//    @Override
//    public boolean isConnect() {
//        return connect;
//    }
//
//    @Override
//    public boolean getConnection() {
//        CoreEquiptAcqConfigBean eptAcqBean = getEptAcqBean();
//        String host = eptAcqBean.getIp();
//        int port = Integer.parseInt(eptAcqBean.getPort());
//        boolean connect = isConnect();
//        if(!connect){
//            getLogger().info(Thread.currentThread().getName() + "---try getConnection for remoteAddress={}",(host+":"+port));
//            //尝试连接一次
//            tryConnect();
//        }else{
//            getLogger().info(Thread.currentThread().getName() + "---reuse getConnection for remoteAddress={}",(host+":"+port));
//        }
//        return isConnect();
//    }
//
//    private void tryConnect() {
//        CoreEquiptAcqConfigBean eptAcqBean = getEptAcqBean();
//        String hostUrl = "tcp://"+eptAcqBean.getIp()+":"+eptAcqBean.getPort();
//        String clientId = eptAcqBean.getExt$Item("clientid");
//        String username = eptAcqBean.getExt$Item("username");
//        String password = eptAcqBean.getExt$Item("password");
//
//        boolean connectFlag = false;
//        try {
//            // 创建客户端
//            if (null == client) {
//                client = new MqttClient(hostUrl, clientId, new MemoryPersistence());
//                client.setManualAcks(false);
//            }
//            // 设置配置
//            MqttConnectOptions options = new MqttConnectOptions();
//            options.setUserName(username);
//            options.setPassword(password.toCharArray());
//            options.setServerURIs(new String[]{hostUrl});
//
//            // 连接超时时间 S
//            options.setConnectionTimeout(15);
//            // 设置会话心跳时间 S
//            options.setKeepAliveInterval(20);
//            // 最多待确认消息的数量 S
////        options.setMaxInflight(30);
//            // 断开时不清除session，重用会话
//            options.setCleanSession(false);
//            //cleanSession 设为 true;当客户端掉线时;服务器端会清除 客户端session;重连后 客户端会有一个新的session,cleanSession
//            // 设为false，客户端掉线后 服务器端不会清除session，当重连后可以接收之前订阅主题的消息。当客户端上线后会接受到它离线的这段时间的消息
//            // 自动重连
//            options.setAutomaticReconnect(true);
//            // 最大重连时长，mill
//            options.setMaxReconnectDelay(60000);
//
//            client.setCallback(new MqttCallbackExtended() {
//                //连接成功回调，需要重新订阅主题
//                @Override
//                public void connectComplete(boolean reconnect, String serverURI) {
//                    setConnect(true);
//                    cacheSender();
//                    subscribe();
//                    waitConnectLatch.countDown();
//                    getLogger().info(Thread.currentThread().getName() + "---connectSuccess--------->address={}", hostUrl);
//                }
//
//                @Override
//                public void connectionLost(Throwable cause) {
//                    getLogger().info(Thread.currentThread().getName() + "---connectionLost--------->address={}", hostUrl);
//                    setConnect(false);
//                    reconnectTimes--;
//                    if (reconnectTimes<0){
//                        close();
//                    }
//                }
//
//                @Override
//                public void deliveryComplete(IMqttDeliveryToken token) {
//                    getLogger().info(Thread.currentThread().getName() + "---deliveryComplete--------->address={}", hostUrl);
//                }
//
//                @Override
//                public void messageArrived(String topic, MqttMessage message) {
//                    if (msg == null){
//                        msg = new HashMap<>();
//                    }
//                    msg.put(topic,new String(message.getPayload(), StandardCharsets.UTF_8));
//                    try {
//                        getMessage(topic, message);
//                    }catch (Exception e){
//                        getLogger().error(e.getMessage());
//                    }
//                }
//            });
//            waitConnectLatch = new CountDownLatch(1);
//            client.connect(options);
//            boolean await = waitConnectLatch.await(waitConnectSecond,TimeUnit.SECONDS);
//            getLogger().info(Thread.currentThread().getName()+"---waitConnectLatch--------->remoteAddress={}",hostUrl);
//        }catch (Exception e){
//            getLogger().error(Thread.currentThread().getName()+"---connectFail--------->remoteAddress={},cause={}",hostUrl,e);
//        }
//    }
//
//    @Override
//    public String send(LimsEquiptAcqConfigInstBean configInstBean) {
//        CoreEquiptAcqConfigBean eptAcqBean = getEptAcqBean();
//        String host = eptAcqBean.getIp();
//        int port = Integer.parseInt(eptAcqBean.getPort());
//        if(configInstBean==null || !isConnect()){
//            getLogger().error(Thread.currentThread().getName()+"---sendFail--------->remoteAddress={},cause={}",host+":"+port,"未连接");
//            close();
//            throw new InvalidDataException("连接已断开，发送指令失败");
//        }
//
//        int qos = Integer.parseInt(configInstBean.getQos());
//        String instruction = configInstBean.getInstruction();
//        if (ObjectUtils.isEmpty(instruction)){
//            throw new InvalidDataException("发送指令为空");
//        }
//        String[] sendMsg = instruction.split(":");
//        String topic = sendMsg[0];
//        String publishMsg = sendMsg[1];
//        MqttMessage message = new MqttMessage();
//        message.setQos(qos);
//        message.setPayload(publishMsg.getBytes());
//        MqttTopic mTopic = client.getTopic(topic);
//        if (null == mTopic) {
//            throw new InvalidDataException("topic：" + topic + " 不存在");
//        }
//        MqttDeliveryToken token = null;
//        try {
//            token = mTopic.publish(message);
//            token.waitForCompletion();
//        } catch (MqttPersistenceException e) {
//            e.printStackTrace();
//        } catch (MqttException e) {
//            e.printStackTrace();
//        }
//        if (token != null && !token.isComplete()) {
//            return "success";
//        }
//        return "fail";
//    }
//
//    public void subscribe() {
//        CoreEquiptAcqConfigBean eptAcqBean = getEptAcqBean();
//        String hostUrl = eptAcqBean.getIp()+":"+eptAcqBean.getPort();
//        if (instMap == null){
//            instMap = new HashMap<>();
//            LimsEquiptAcqConfigInstService limsEquiptAcqConfigInstService = ApplicationContextHelper.getBean(LimsEquiptAcqConfigInstService.class);
//            List<LimsEquiptAcqConfigInstBean> instBeans = limsEquiptAcqConfigInstService.selectListByFilter(SearchFilter.instance()
//                    .match("acqConfigId",eptAcqBean.getId()).filter(MatchPattern.EQ)
//                    .match("conversationRole","subscribe").filter(MatchPattern.EQ));
//            for (LimsEquiptAcqConfigInstBean instBean : instBeans){
//                instMap.put(instBean.getInstruction(),instBean);
//            }
//        }
//        try {
//            if (!ObjectUtils.isEmpty(instMap)) {
//                String[] topics = instMap.keySet().toArray(new String[instMap.size()]);
//                int[] qos = instMap.values().stream().mapToInt(e -> Integer.parseInt(e.getQos())).toArray();
//                client.subscribe(topics, qos);
//            }
//        }catch (Exception e){
//            getLogger().error(Thread.currentThread().getName()+"---subscribeFail--------->remoteAddress={},cause={}",hostUrl,e);
//        }
//    }
//
//    private void getMessage(String topic, MqttMessage message){
//        String messg = new String(message.getPayload(), StandardCharsets.UTF_8);
//        CoreEquiptAcqConfigBean eptAcqBean = getEptAcqBean();
//        String host = eptAcqBean.getIp();
//        int port = Integer.parseInt(eptAcqBean.getPort());
//        String key = host+":"+port;
//
//        getLogger().info("为设备={},接受数据.........", key);
//        LimsEquiptAcqConfigInstBean configInstBean = getInstBeanByTopic(topic);
//        String resultSaveType = "";
//        if (!ObjectUtils.isEmpty(configInstBean)) {
//            LimsEquiptAcqScheduleTaskBean taskBean = new LimsEquiptAcqScheduleTaskBean();
//            taskBean.setCreatedTime(LocalDateTime.now());
//            try {
//                if (configInstBean != null) {
//                    getLogger().info("设备={},数据={},正在接收数据.........", key,messg);
//                    resultSaveType = configInstBean.getResultSaveType();
//                    taskBean.setAcqType(configInstBean.getInstructionType());
//                    taskBean.setEquiptId(configInstBean.getAcqConfigId());
//                    taskBean.setAcqData(messg);
//                    taskBean.setResult("success");
//                }
//            } catch (Exception e) {
//                e.printStackTrace();
//                getLogger().error("连接={}, 数据={},接收数据失败..........", key, configInstBean.getInstruction());
//                taskBean.setResult("fail");
//                taskBean.setErrorMsg(e.getMessage());
////                    setConnect(false);
//            }
//            //根据数据存储方式不一样，做不同的存储操作
//            if ("allData".equals(resultSaveType)) {
//                taskBean.setId(ApplicationContextHelper.getNextIdentity());
//                ApplicationContextHelper.getBean(LimsEquiptAcqScheduleTaskServiceImpl.class).getDao().insert(taskBean);
//            } else {
//                RedisHelper.put("EQUIPT_RETURN_DATA", configInstBean.getId().toString(), taskBean);
//            }
//        }
//        Thread thread = Thread.currentThread();
//        getLogger().error( "{},{}---read interrupt--------->remoteAddress={}",thread.getId(),thread.getName(),key);
//    }
//
//    @Override
//    public void close() {
//        getLogger().error("-----close---------");
//        getLogger().info("为设备={},,close",getKey(getEptAcqBean()));
//        try {
//            client.disconnect();
//        }catch (Exception e){
//            e.printStackTrace();
//        }finally {
//            try{
//                client.close();
//            }catch (Exception e){
//                e.printStackTrace();
//            }
//            setConnect(false);
//        }
//    }
//
//    public LimsEquiptAcqConfigInstBean getInstBeanByTopic(String topic){
//        return instMap.get(topic);
//    }
//}
